package com.heima.schedule.service;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.heima.common.cache.CacheService;
import com.heima.common.constants.ScheduleConstants;
import com.heima.model.schedule.dto.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    private TaskinfoService taskinfoService;

    @Autowired
    private TaskinfoLogsService taskinfoLogsService;

    @Autowired
    private CacheService cacheService;

    /**
     * 添加任务
     *
     * @param task 任务数据
     * @return 返回任务结果id
     */
    @Override
    @Transactional
    public Long addTask(Task task) {

        // 0 效验task是否为空
        if (task == null) {
            log.warn("入参不能为空");
            return null;
        }

        // 1. 插入taskinfo表
        Taskinfo taskInfo = new Taskinfo();
        BeanUtils.copyProperties(task, taskInfo);
        taskInfo.setExecuteTime(new Date(task.getExecuteTime()));
        boolean taskInfoResult = taskinfoService.save(taskInfo);
        if (!taskInfoResult) {
            log.warn("任务插入失败");
            return null;
        }

        // 2. 插入taskinfolog表
        TaskinfoLogs logs = new TaskinfoLogs();
        BeanUtils.copyProperties(task, logs);
        logs.setExecuteTime(new Date(task.getExecuteTime()));
        boolean logsResult = taskinfoLogsService.save(logs);
        if (!logsResult) {
            log.warn("日志插入失败");
            throw new RuntimeException("日志插入失败");
        }

        // 3. 判断任务执行时间是否小于当前时间
        // 任务执行时间
        long executeTime = task.getExecuteTime();

        // 当前时间
        long currentTime = System.currentTimeMillis();
        if (executeTime <= currentTime) {
            // 加入到当前执行队列
            cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));

        }

        // 4. 判断任务执行时间是否小于未来五分钟
        long futureTime = currentTime + (5 * 60 * 1000);
        if (executeTime <= futureTime) {
            cacheService.zAdd("FUTUR", JSON.toJSONString(task), executeTime);
        }
        // 5. 返回任务id

        return taskInfo.getTaskId();
    }

    /**
     * 取消任务
     *
     * @param taskId 传入任务id
     * @return 取消任务的结果
     */
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public boolean cancelTask(long taskId) {

        // 1. 删除数据库任务列表中的任务
        boolean removeResult = taskinfoService.removeById(taskId);
        if (!removeResult) {
            log.warn("删除任务失败");
            return false;
        }

        // 2. 查询数据库任务记录列表
        TaskinfoLogs taskinfoLogs = taskinfoLogsService.getById(taskId);

        // 3. 更新任务记录列表中的状态(已取消状态)
        taskinfoLogs.setStatus(ScheduleConstants.CANCELLED);
        boolean updateResult = taskinfoLogsService.updateById(taskinfoLogs);
        if (!updateResult) {
            log.warn("任务记录更新失败");
            throw new RuntimeException("任务记录更新失败");
        }

        // 4. 组装task数据
        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs, task);
        // 日期转换
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());

        // 如果小于等于当前时间，删除TOPIC执行列队中任务，否则删除FUTURE未来列队中的任务
        if (task.getExecuteTime() <= System.currentTimeMillis()) {
            cacheService.lRemove(ScheduleConstants.TOPIC, 0, JSON.toJSONString(task));
        } else {
            cacheService.zRemove(ScheduleConstants.FUTURE, JSON.toJSONString(task));
        }
        return true;
    }

    /**
     * 消费任务
     *
     * @return 消费结果
     */
    @Override
    @Transactional(rollbackFor = RuntimeException.class)
    public boolean pollTask() {

        // 1. 从TOPIC中弹出数据
        String taskString = cacheService.lRightPop("TOPIC");
        if (StringUtils.isEmpty(taskString)) {
            log.warn("没有可执行任务");
            return false;
        }

        // 2.将task结构的JSON字符串转换为TASK对象
        Task task = JSON.parseObject(taskString, Task.class);
        if (task == null) {
            log.warn("没有可执行任务");
            return false;
        }

        // 3. 删除数据库任务 注意：此处是对Redis的操作，无法回调 (解决方案，重试机制try{}catch(){} )
        boolean taskInfoResult = taskinfoService.removeById(task.getTaskId());
        if (!taskInfoResult) {
            log.warn("删除任务失败");
            return false;
        }

        // 4. 更新数据库任务记录表中的状态
        TaskinfoLogs logs = new TaskinfoLogs();
        BeanUtils.copyProperties(task, logs);

        // 时间类型转换
        logs.setExecuteTime(new Date(task.getExecuteTime()));
        // 设置状态
        logs.setStatus(1);
        boolean logsResult = taskinfoLogsService.updateById(logs);
        if (!logsResult) {
            log.warn("日志更新失败");
            throw new RuntimeException("日志更新失败");
        }
        return true;
    }

    /**
     * 定时器(刷新)
     */
    @Scheduled(fixedRate = 1000 * 60 * 5)
    public void refresh() {
        log.warn("定时器被执行了");

        // 分布式锁
        String lockTonken = cacheService.tryLock("refresh_lock", 5000);

        // 如果锁为空，则返回
        if (StringUtils.isEmpty(lockTonken)) {
            return;
        }

        // 获取未来消息列队中的数据(按分值获取)
        Set<String> future = cacheService.zRangeByScore("FUTURE", 0, System.currentTimeMillis());
        if (CollectionUtils.isEmpty(future)) {
            log.warn("没有可移动任务");
            return;
        }

        // 将任务从未来队列转移到执行队列
        cacheService.refreshWithPipeline("FUTURE", "TOPIC", future);
    }

    /**
     * 初始化数据
     */
    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void initDate() {

        // 0. 清除Redis里面的数据
        clear();

        // 1. 读取MySQL里面的任务
        List<Taskinfo> list = taskinfoService.list();
        if (CollectionUtils.isEmpty(list)) {
            log.warn("任务为空");
            return;
        }

        // 2. 判断任务时间，进行分配
        for (Taskinfo taskinfo : list) {
            if (taskinfo == null) {
                continue;
            }

            // 任务执行时间
            long executeTime = taskinfo.getExecuteTime().getTime();
            // 当前时间
            long currentTime = System.currentTimeMillis();
            // 未来时间
            long futureTime = currentTime + (5 * 60 * 1000);

            Task task = new Task();
            BeanUtils.copyProperties(taskinfo, task);
            task.setExecuteTime(executeTime);

            if (executeTime <= currentTime) {
                cacheService.lLeftPush("TOPIC", JSON.toJSONString(task));
            } else if (executeTime <= futureTime) {
                cacheService.zAdd("FUTURE", JSON.toJSONString(task), executeTime);
            }
        }

    }

    /**
     * 清空TOPIC和FUTURE中的任务
     */
    private void clear() {
        cacheService.delete("TOPIC");
        cacheService.delete("FUTURE");
    }

}
